package org.lemon.transport.context.service.request;

import com.google.gson.Gson;
import com.google.gson.JsonObject;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.lemon.transport.context.data.device.DeviceProfile;
import org.lemon.transport.context.data.message.MessageMetaData;
import org.lemon.transport.context.data.message.ToRuleEngineMessage;
import org.lemon.transport.context.data.tenant.TenantId;
import org.lemon.transport.context.service.TransportDeviceProfileCache;
import org.lemon.transport.context.service.TransportServiceCallback;
import org.lemon.transport.context.service.queue.*;
import org.lemon.transport.context.service.queue.discovery.HashPartitionService;
import org.lemon.transport.context.data.queue.message.ProtoQueueMessage;
import org.lemon.transport.context.service.queue.provider.RabbitMqTransportQueueFactory;
import org.lemon.transport.context.service.queue.stats.MessagesStats;
import org.lemon.transport.context.service.queue.stats.StatsFactory;
import org.lemon.transport.context.service.queue.stats.StatsType;
import org.lemon.transport.context.session.SessionMsgType;
import org.server.transport.gen.SessionProto;
import org.server.transport.gen.TransportProto;
import org.springframework.stereotype.Component;

import java.util.UUID;

@Getter
@Slf4j
@Component
public class DeviceTransportRequestTemplate {

    private final Gson gson = new Gson();

    private MessagesStats ruleEngineProducerStats;
    private MessagesStats coreProducerStats;
    private MessagesStats transportApiStats;

    private TransportDeviceProfileCache deviceProfileCache;
    private HashPartitionService partitionService;

    /**
     * 消息请求api模板
     */
    protected QueueRequestTemplate<ProtoQueueMessage<TransportProto.TransportApiRequestMsg>, ProtoQueueMessage<TransportProto.TransportApiResponseMsg>> transportApiRequestTemplate;
    /**
     * 发送至规则引擎的消息生产者
     */
    protected QueueProducer<ProtoQueueMessage<TransportProto.ToRuleEngineMsg>> ruleEngineMessageProducer;
    /**
     * 发送至应用服务的消息生产者
     */
    protected QueueProducer<ProtoQueueMessage<TransportProto.ToCoreMsg>> coreMessageProducer;

    public DeviceTransportRequestTemplate(RabbitMqTransportQueueFactory queueFactory,
                                          StatsFactory statsFactory,
                                          TransportDeviceProfileCache deviceProfileCache) {

        this.ruleEngineProducerStats = statsFactory.createMessagesStats(StatsType.RULE_ENGINE.getName() + ".producer");
        this.coreProducerStats = statsFactory.createMessagesStats(StatsType.CORE.getName() + ".producer");
        this.transportApiStats = statsFactory.createMessagesStats(StatsType.TRANSPORT.getName() + ".producer");

        this.transportApiRequestTemplate = queueFactory.createTransportApiRequestTemplate();
        this.ruleEngineMessageProducer = queueFactory.createRuleEngineMessageProducer();
        this.coreMessageProducer = queueFactory.createCoreMessageProducer();

        this.transportApiRequestTemplate.setMessagesStats(transportApiStats);

        this.deviceProfileCache = deviceProfileCache;
    }

    private void sendToRuleEngine(TenantId tenantId, Long deviceId, Long projectId, SessionProto.SessionInfoProto sessionInfo, JsonObject json,
                                  MessageMetaData metaData, SessionMsgType sessionMsgType, QueueCallback callback) {
        DeviceProfile deviceProfile = deviceProfileCache.get(sessionInfo.getDeviceProfileId());
        Long ruleChainId;
        String queueName;

        if (deviceProfile == null) {
            log.warn("[{}] Device profile is null!", sessionInfo.getDeviceProfileId());
            ruleChainId = null;
            queueName = null;
        } else {
            ruleChainId = deviceProfile.getDefaultRuleChainId();
            queueName = deviceProfile.getDefaultQueueName();
        }
        ToRuleEngineMessage message = ToRuleEngineMessage.newMessage(queueName, sessionMsgType.name(), String.valueOf(deviceId), projectId, metaData, gson.toJson(json), ruleChainId, null);
        sendToRuleEngine(tenantId, message, callback);
    }

    protected void sendToDeviceActor(SessionProto.SessionInfoProto sessionInfo, TransportProto.TransportToDeviceActorMsg toDeviceActorMsg, TransportServiceCallback<Void> callback) {

        TopicPartitionInfo tpi = partitionService.resolve(ServiceType.CORE, getTenantId(sessionInfo), sessionInfo.getDeviceId());
        if (log.isTraceEnabled()) {
            log.trace("[{}][{}] Pushing to topic {} message {}", getTenantId(sessionInfo), sessionInfo.getDeviceId(), tpi.getFullTopicName(), toDeviceActorMsg);
        }
        TransportQueueCallback transportTbQueueCallback = callback != null ?
                new TransportQueueCallback(callback) : null;
        coreProducerStats.incrementTotal();
        TransportProto.ToCoreMsg msg = TransportProto.ToCoreMsg.newBuilder()
                .setToDeviceActorMsg(toDeviceActorMsg)
                .build();
        ProtoQueueMessage<TransportProto.ToCoreMsg> queueMessage = new ProtoQueueMessage<TransportProto.ToCoreMsg>(UUID.fromString(String.valueOf(getRoutingKey(sessionInfo))), msg);
        StatsCallback wrappedCallback = new StatsCallback(transportTbQueueCallback, coreProducerStats);
        coreMessageProducer.send(tpi, queueMessage, wrappedCallback);
    }

    private void sendToRuleEngine(TenantId tenantId, ToRuleEngineMessage message, QueueCallback callback) {

        TopicPartitionInfo tpi = partitionService.resolve(ServiceType.RULE_ENGINE, message.getQueueName(), tenantId, Long.valueOf(message.getOriginator()));
        if (log.isTraceEnabled()) {
            log.trace("[{}][{}] Pushing to topic {} message {}", tenantId, message.getOriginator(), tpi.getFullTopicName(), message);
        }
        TransportProto.ToRuleEngineMsg msg = TransportProto.ToRuleEngineMsg.newBuilder()
                .setMsg(ToRuleEngineMessage.toByteString(message))
                .setTenantId(tenantId.getValue())
                .build();
        ruleEngineProducerStats.incrementTotal();
        StatsCallback wrappedCallback = new StatsCallback(callback, ruleEngineProducerStats);
        ruleEngineMessageProducer.send(tpi, new ProtoQueueMessage<>(message.getId(), msg), wrappedCallback);
    }

    protected Long getRoutingKey(SessionProto.SessionInfoProto sessionInfo) {
        return sessionInfo.getDeviceId();
    }

    protected TenantId getTenantId(SessionProto.SessionInfoProto sessionInfo) {
        return TenantId.valueOf(sessionInfo.getTenantId());
    }

    private class TransportQueueCallback implements QueueCallback {

        private final TransportServiceCallback<Void> callback;

        private TransportQueueCallback(TransportServiceCallback<Void> callback) {
            this.callback = callback;
        }

        @Override
        public void onSuccess() {

        }

        @Override
        public void onFailure(Throwable t) {
        }
    }

    private static class StatsCallback implements QueueCallback {
        private final QueueCallback callback;
        private final MessagesStats stats;

        private StatsCallback(QueueCallback callback, MessagesStats stats) {
            this.callback = callback;
            this.stats = stats;
        }

        @Override
        public void onSuccess() {
            stats.incrementSuccessful();
            if (callback != null)
                callback.onSuccess();
        }

        @Override
        public void onFailure(Throwable t) {
            stats.incrementFailed();
            if (callback != null)
                callback.onFailure(t);
        }
    }
}
